package com.itcast.flink.connectors.jdbc;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * <p>Description: </p>
 *
 * @author
 * @version 1.0
 * <p>Copyrigh:Copyright(c)2020</p>
 * @date
 */
public class MySQLSinkFunction extends RichSinkFunction<AccessLog> {
    
    /**
     * 数据库连接
     */
    private Connection connection;
    
    /**
     * 预编译SQL
     */
    private PreparedStatement preparedStatement;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 建立Mysql连接
        String jdbcUrl = "jdbc:mysql://192.168.23.140:3306/flink?useSSL=false";
        String executeSql = "insert into t_access_log (ip, time, type, api) values(?, ?, ?, ?)";
        connection = DriverManager.getConnection(jdbcUrl, "root", "654321");
        preparedStatement = connection.prepareStatement(executeSql);
    }
    
    @Override
    public void close() throws Exception {
        // 释放数据库连接
        if (null != connection) {
            try {
                connection.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
            connection = null;
        }
    }
    
    @Override
    public void invoke(AccessLog value, Context context) throws Exception {
        // 执行写入数据
        preparedStatement.setString(1, value.getIp());
        preparedStatement.setString(2, value.getTime());
        preparedStatement.setString(3, value.getType());
        preparedStatement.setString(4, value.getApi());
        preparedStatement.executeUpdate();
        
    }
}
